
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, Duration, StreamingContext}

/**
 * Created by chenjianwen on 2016/2/24.
 * spark streaming wordcount例子
 */
object wordcount {
  def main (args: Array[String]) {
    //指定master的url
    val conf = new SparkConf().setMaster("spark://master:7077").setAppName("haha")
    //参数二：指定批处理的时间片为1秒(每一秒都会生成一个DStream并提交到Spark执行)
    val ssc = new StreamingContext(conf,Seconds(1))

    //输入流的ip和端口，每一行都会有一个DStream产生
    val lines = ssc.socketTextStream("master",9999)

    //下 面的操作和RDD的操作基本一致
    val words = lines.flatMap(_.split("\\s"))
    val pair = words.map(x=>(x,1))
    val allRes = pair.reduceByKey(_+_)

    //相当于RDD的Action算子，触发一个DStream的RDD DAG的执行。
    allRes.print()
    //启动
    ssc.start()
    ssc.awaitTermination()
  }
}
